package com.shujia.dwd

import com.shujia.util.SparkTool
import org.apache.spark.sql._

/**
  *
  * 位置融合表
  * 提取OIDD、移动DPI和WCDR分钟粒度位置数据，对重复时间点数据进行二次清洗，完成后写入合并文件库；根据乒乓切换、超速数据等清洗原则对噪声数据进行清洗，完成后写入融合文件库；移动DDR数据在数据上传后延迟一天进行。
  *
  * 将原始信令数据融合处理后，根据应用的需求和后期应用场景的需求，共保留12个字段。包括手机号、城市代码、业务开始时间、经纬度、BSID、对端号码、数据源和业务类型。
  *
  */
object MergeLocationApp extends SparkTool {


  /**
    * 在子类中实现run方法，实现自定义的代码累哦及
    *
    * import spark.implicits._
    * import org.apache.spark.sql.functions._
    *
    * @param spark spark的环境
    */
  override def run(spark: SparkSession): Unit = {

    import spark.implicits._

    //$day_id  scala 语言的字符串的拼接
    val ddr: DataFrame = spark.sql(s"select * from ods.ods_ddr where day_id=$day_id")
    val dpi: DataFrame = spark.sql(s"select * from ods.ods_dpi where day_id=$day_id")
    val oidd: DataFrame = spark.sql(s"select * from ods.ods_oidd where day_id=$day_id")
    val wcdr: DataFrame = spark.sql(s"select * from ods.ods_wcdr where day_id=$day_id")


    /**
      * 对四类数据进行合并得到位置融合表
      *
      */
    val unionDF: Dataset[Row] = ddr.union(dpi).union(oidd).union(wcdr)

    /**
      *
      * mdn string comment '手机号码'
      * ,start_time string comment '业务时间'
      * ,county_id string comment '区县编码'
      * ,longi string comment '经度'
      * ,lati string comment '纬度'
      * ,bsid string comment '基站标识'
      * ,grid_id string comment '网格号'
      * ,biz_type string comment '业务类型'
      * ,event_type string comment '事件类型'
      * ,data_source string comment '数据源'
      *
      * 对用户敏感信息进行脱敏，
      * 使用md5加密
      *
      */

    val resultDF: DataFrame = unionDF.select(
      hashMd5($"mdn") as ",md5",
      $"start_time",
      $"county_id",
      $"longi",
      $"lati",
      $"bsid",
      $"grid_id",
      $"biz_type",
      $"event_type",
      $"data_source"
    )


    /**
      * 保存数据到融合表中
      *
      */

    saveDataAndAddPartitionWithDay(
      resultDF,
      "/daas/motl/dwd/dwd_res_regn_mergelocation_msk_d",
      "dwd.dwd_res_regn_mergelocation_msk_d "
    )

  }
}
